package bch.rocketmq.consumer;

import bch.rocketmq.constant.MqConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.MQPullConsumerScheduleService;
import org.apache.rocketmq.client.impl.consumer.ConsumeMessageOrderlyService;
import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
import org.apache.rocketmq.client.impl.consumer.PullMessageService;
import org.apache.rocketmq.client.impl.consumer.PullRequest;
import org.apache.rocketmq.client.impl.factory.MQClientInstance;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

@Slf4j
@Component
public class MQListener {

    // 就是DefaultMQPushConsumer, 就是升级版的拉模式消费者
    /**
     * {@link ConsumeMessageOrderlyService.ConsumeRequest#run() 顺序消费}
     */

    /**
     * {@link MQPullConsumerScheduleService}
     *
     * {@link MQClientInstance#start() 启动拉取定时器 心跳 重平衡... 定时器}
     * {@link PullMessageService}
     * {@link DefaultMQPushConsumerImpl#pullMessage(PullRequest)}
     *
     * {@link org.apache.rocketmq.spring.annotation.ExtRocketMQConsumerConfiguration} 扩展、可自定义、适合复杂场景
     */
    @Service
    @RocketMQMessageListener(topic = MqConstant.TOPIC_NAME, selectorExpression = "tag1", consumerGroup = "gr oup1", consumeMode = ConsumeMode.ORDERLY)
    public static class Consumer1 implements RocketMQListener<String> {
        @Override
        public void onMessage(String msg) {
            log.info("Consumer1 监听到消息：{}", msg);
        }
    }

    @Service
    @RocketMQMessageListener(topic = MqConstant.TOPIC_NAME, selectorExpression = "tag1", consumerGroup = "group2", consumeMode = ConsumeMode.ORDERLY)
    public static class Consumer2 implements RocketMQListener<MessageExt> {
        @Override
        public void onMessage(MessageExt messageExt) {
            byte[] body = messageExt.getBody();
            String msg = new String(body);
            log.info("Consumer2 监听到消息：{}", msg);
        }
    }

}
